package com.ndp.fb.fbagent.service.impl;

import com.alibaba.dubbo.config.annotation.Service;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.ndp.fb.config.ConfigCentre;
import com.ndp.fb.constant.ConfigConst;
import com.ndp.fb.fbagent.service.RedisParamsFBService;
import com.ndp.fb.fbagent.service.help.CheckAccessTokenHelper;
import com.ndp.fb.fbagent.service.webrequest.QuartzWebRequestor;
import com.ndp.fb.model.fbagent.param.ApiCallBackTarget;
import com.ndp.fb.model.fbagent.param.CallBackModel;
import com.ndp.fb.model.fbagent.result.stats.FBAdStats;
import com.ndp.fb.mq.api.producer.RocketMqProducerService;
import com.ndp.fb.redis.api.RedisClient;
import com.ndp.fb.redis.model.RedisParams;
import com.ndp.fb.util.StringUtil;
import com.ndp.fb.util.ThreadPool;
import com.ndp.fb.util.UUID;
import com.restfb.BinaryAttachment;
import com.restfb.DefaultJsonMapper;
import com.restfb.JsonMapper;
import com.restfb.WebRequestor;
import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.annotation.Autowired;

import java.io.IOException;
import java.util.*;
import java.util.concurrent.TimeUnit;

import static java.net.HttpURLConnection.HTTP_NOT_MODIFIED;
import static java.net.HttpURLConnection.HTTP_OK;

/**
 * 处理quartz发起的对fb的请求代码
 * Created by infi on 2015/1/22.
 */
@Service
public class RedisParamsFBServiceImpl implements RedisParamsFBService, InitializingBean {

    private static final Logger logger = LoggerFactory.getLogger(RedisParamsFBServiceImpl.class);


    private static final String STOP_REQEUST_PREFIX = "error_stop_act_";
    private static final String LIMITED_PREFIX = "limited_act_";
    private static final int limitControlTimeOut = 330 * 1000; //超限时间
    private static final int DEFAULT_RETRY_TIMES = 3; //重试
    //90000001 IOException network exception
    //1487390 Service temporarily unavailable
    private List<String> retryErrorCodeList = Arrays.asList("90000001", "1487390");
    private List<String> stopSendCodeList = Arrays.asList("10", "274", "190");
    private int stopSendCodeTimeOut = 2;
    private List<String> callLimitedCodeList = Arrays.asList("1487742");
    private JsonMapper jsonMapper = new DefaultJsonMapper();

    @Autowired
    private RocketMqProducerService rocketMqProducerService;
    @Autowired
    private RedisClient redisClient;
    @Autowired
    CheckAccessTokenHelper checkAccessTokenHelper;

    @Override
    public void afterPropertiesSet() throws Exception {
        String stopSendCodeString = ConfigCentre.getString(ConfigConst.ERROR_STOP_ERROR_CODE);
        if (StringUtils.isNoneEmpty(stopSendCodeString)) {
            stopSendCodeList = Arrays.asList(stopSendCodeString.split("#"));
            stopSendCodeTimeOut = ConfigCentre.getInteger(ConfigConst.ERROR_STOP_TIMEOUT_MINUTES);
        } else {
            logger.error("没有配置error.stop.error.code,将使用默认值");
        }
    }

    @Override
    public String makeRequest(RedisParams redisParams) throws IOException {
        Long start = System.currentTimeMillis();
        logger.info("RedisParamsFBServiceImpl makeRequest 消费请求 >>>> " + redisParams.getUuid() + "url>>>>" + redisParams.getUrl() + ",start >>>> " + start);
        String result = checkAccessTokenHelper.checkAccessToken(redisParams);//检查 token是否失效
        String body = null;
        if (StringUtils.isEmpty(result)) {
            body = getResponse(redisParams).getBody();
        } else {
            body = result;
        }
        Long end = System.currentTimeMillis();
        logger.info("RedisParamsFBServiceImpl makeRequest 消费请求完成 >>>> " + redisParams.getUuid() + "url>>>>" + redisParams.getUrl() + " ,耗时 >>>" + (end - start));
        return body;
    }

    @Override
    public void makeRequestAsyn(final RedisParams redisParams) {
        ThreadPool.execute(new Runnable() {
            @Override
            public void run() {
                getResponseAsyn(redisParams);
            }
        });
    }

    private void getResponseAsyn(RedisParams redisParams) {
        Long start = System.currentTimeMillis();
        logger.info("makeRequestAsyn 消费请求 >>>> " + redisParams.getUuid() + ",start >>>> " + start);
        String body = null;
        int statusCode = HTTP_OK;
        String accountId = redisParams.getAccountId();
        String targetAndEquipment = redisParams.getTargetAndEquipment();
        String etagKey = null;
        if (StringUtils.isNotEmpty(targetAndEquipment)) {
            CallBackModel callBackModel = JSON.parseObject(targetAndEquipment, CallBackModel.class);
            etagKey = callBackModel.getEquipment() == null ? null : callBackModel.getEquipment().get("etagKey");
        }
        try {
            String result = checkAccessTokenHelper.checkAccessToken(redisParams);//检查当前此token是否失效
            if (StringUtils.isEmpty(result)) {
                WebRequestor.Response response = getResponse(redisParams);
                if (null == response) {
                    logger.error("fb resule >>>> response is null");
                    body = "{\"error\":\"fb resule >>>> response is null\",\"error_subcode\":90000000}";
                } else {
                    body = response.getBody();
                    statusCode = response.getStatusCode();
                }
            } else {
                body = result;
            }
        } catch (IOException e) {
            body = "{\"error\":{\"message\":\" 请求fb异常IOException," + e.getMessage() + "\",\"error_subcode\":90000001}}";
            logger.error(body, e);
        } catch (Throwable e) {
            body = "{\"error\":{\"message\":\" 请求fb异常Throwable," + e.getMessage() + "\",\"error_subcode\":90000002}}";
            logger.error(body, e);
        } finally {
            if (HTTP_NOT_MODIFIED == statusCode && StringUtils.isNotEmpty(etagKey)) {  // 如果是通过eTag查询并且返回结果是304，则不用再进行后续处理
                logger.info("ETag查询返回值未发生变化>>>url" + redisParams.getUrl() + ", etagKey>>>" + etagKey);
            } else {
                JSONObject jsonObject = null;
                try {
                    jsonObject = JSON.parseObject(body);
                    if (jsonObject.containsKey("error")) {
                        JSONObject errorObject = jsonObject.getJSONObject("error");
                        String code = errorObject.getString("code");
                        String errorSubCode = errorObject.getString("error_subcode");
                        String message = errorObject.getString("message");
                        String errorType = errorObject.getString("type");

                        //是否需暂停某个账号下面的所有广告
                        if (stopSendCodeList.contains(code)) {
                            logger.error("facebook error code " + code + "  , accountId >>>> " + accountId);
                            redisClient.set(STOP_REQEUST_PREFIX + accountId, "1");
                            redisClient.expire(STOP_REQEUST_PREFIX + accountId, stopSendCodeTimeOut, TimeUnit.MINUTES.name());
                            logger.info("请求fb出现errorcode>>>> " + code + " ; 停止标记保存redis完成!");
                            sendMessage(redisParams, body);
                            // 检查一下token是否真的失效了,失效了就把数据库置为失效
                            if ("460".equals(errorSubCode)) {// 190-460 错误信息 : "message":"Error validating access token: The session has been invalidated because the user has changed the password.", "type":"OAuthException","code":190,"error_subcode":460
                                String inputToken = checkAccessTokenHelper.getAccessToken(redisParams);
                                checkAccessTokenHelper.checkFBAccessToken(inputToken);
                            }
                        }

                        //是否需要重试，不会sendMessage结果
                        else if (isNeedTry(errorSubCode, errorType, message, redisParams.getUuid(), redisParams.getRetryTimes())) {
                            logger.warn("请求重试 errorSubCode>>>>> " + errorSubCode + " ;body >>>> " + body + " >>>> 需要重发 tmpRetry >>>" + redisParams.getRetryTimes());
                            int tmpRetry = redisParams.getRetryTimes() + 1;
                            redisParams.setRetryTimes(tmpRetry);
                            redisClient.zadd("score_act_" + accountId, redisParams.getScore(), redisParams);
                        }

                        //是否超限需要需要sendMessage结果
                        else if (callLimitedCodeList.contains(errorSubCode)) {
                            redisClient.set(LIMITED_PREFIX + accountId, "1");
                            redisClient.expire(LIMITED_PREFIX + accountId, limitControlTimeOut, TimeUnit.MILLISECONDS.name());
                            sendMessage(redisParams, body);
                        }

                        //别的错误需要sendMessage结果
                        else {
                            errorObject.put("message", message + "___ retryTimes " + redisParams.getRetryTimes());
                            sendMessage(redisParams, jsonObject.toJSONString());
                        }
                    } else {
                        //没有异常
                        cacheEtagData(redisParams.getUrl(), body);
                        try {
                            String resultClass = redisParams.getResultClassName();
                            logger.debug("请求返回值转化类型名称>>" + resultClass);
                            if (StringUtils.isEmpty(resultClass)) {
                                sendMessage(redisParams, body);
                            } else {
                                Class<?> aClass = Class.forName(redisParams.getResultClassName());

                                Object object;
                                try {
                                    object = jsonMapper.toJavaList(body, aClass);
                                } catch (Exception e) {
                                    object = jsonMapper.toJavaObject(body, aClass);
                                }

                                if (object != null) {
                                    sendMessage(redisParams, object);
                                } else {
                                    sendMessage(redisParams, body);
                                }
                            }
                        } catch (ClassNotFoundException ex) {
                            logger.error("返回值转化类型名称找不到 " + redisParams.getResultClassName() + "=================", ex);
                            sendMessage(redisParams, body);
                        } catch (Throwable e) {
                            logger.error("返回值转化类型出现异常", e);
                            sendMessage(redisParams, body);
                        }
                    }
                } catch (Exception e) {
                    //处理异常不维护
                    logger.error("解析fb返回值出现异常 body >>>>> " + body, e);
                    sendMessage(redisParams, body);
                }
            }
        }
        Long end = System.currentTimeMillis();
        logger.info("makeRequestAsyn 消费请求完成 >>>> " + redisParams.getUuid() + " ,耗时 >>>" + (end - start));
    }


    /**
     * 检查是否需要重试
     *
     * @return
     */
    private boolean isNeedTry(String errorSubCode, String message, String errorType, String uuid, int retryTimes) {
        boolean flag = false;
        if ("OAuthException".equals(errorType) && "An unexpected error has occurred. Please retry your request later.".equals(message) && retryTimes < DEFAULT_RETRY_TIMES) {
            logger.debug("uuid >>>>> " + uuid + " ; An unexpected error has occurred. Please retry your request later.错误 >>>> 需要重发 tmpRetry >>>" + retryTimes);
            flag = true;
        } else if ("OAuthException".equals(errorType) && "(#2) Service temporarily unavailable".equals(message) && retryTimes < DEFAULT_RETRY_TIMES) {
            logger.debug("uuid >>>>> " + uuid + " ;(#2) Service temporarily unavailable 错误 >>>> 需要重发 tmpRetry >>>" + retryTimes);
            flag = true;
        } else if (errorSubCode != null && retryErrorCodeList.contains(errorSubCode) && retryTimes < DEFAULT_RETRY_TIMES) {
            logger.debug("uuid >>>>> " + uuid + " ;errorSubCode 错误 >>>> " + errorSubCode + ",需要重发 tmpRetry >>>" + retryTimes);
            flag = true;
        }
        return flag;
    }

    private void cacheEtagData(String url, String body) {
        //etag
    }

    /**
     * 请求fb获取response
     *
     * @param req
     * @return
     * @throws java.io.IOException
     */
    private WebRequestor.Response getResponse(RedisParams req) throws IOException {
        WebRequestor.Response response = null;
        QuartzWebRequestor requester = new QuartzWebRequestor();
        requester.setCheckAccessTokenHelper(checkAccessTokenHelper);
        String url = req.getUrl();
        requester.setAccountId(String.valueOf(req.getAccountId()));
        // Added for ETag
        String targetAndEquipment = req.getTargetAndEquipment();
        String etagKey = null;
        if (StringUtils.isNotEmpty(targetAndEquipment)) {
            CallBackModel callBackModel = JSON.parseObject(targetAndEquipment, CallBackModel.class);
            Map<String, String> equipment = callBackModel.getEquipment();
            if(equipment != null && StringUtil.isNotEmpty(equipment.get("etagKey")))
                etagKey = equipment.get("etagKey");
        }
        if (StringUtils.isNotEmpty(etagKey)) {
            logger.debug("etagKey->" + etagKey);
            requester.setEtagKey(etagKey);
            requester.setRedisClient(redisClient);
        }
        logger.debug("UUID->" + req.getUuid());
        logger.debug("url->" + url);
        logger.debug("method->" + req.getMethodName());
        logger.debug("RetryTimes->" + req.getRetryTimes());
//        logger.debug("TargetAndEquipment->" + req.getTargetAndEquipment());
        if ("get".equalsIgnoreCase(req.getMethodName())) {
            response = requester.executeGet(url);
        } else if ("post".equalsIgnoreCase(req.getMethodName())) {
            String fileName = req.getFileName();
            byte[] context = req.getContext();
            if (fileName != null && context != null) {
                BinaryAttachment[] binaryAttachments = new BinaryAttachment[1];
                binaryAttachments[0] = BinaryAttachment.with(fileName, context);
                response = requester.executePost(url, req.getParams(), binaryAttachments);
            } else {
                response = requester.executePost(url, req.getParams(), null);
            }
        } else {
            logger.error("http请求method错误 >>>> " + req.getMethodName());
        }
        return response;
    }

    /**
     * 处理结果
     *
     * @param req
     * @param body
     */
    private void sendMessage(RedisParams req, Object body) {
        logger.debug("xxxxxxx uuid >>>>  " + req.getUuid());
        String targetAndEquipment = req.getTargetAndEquipment();
        if (targetAndEquipment == null) {
            redisClient.set(req.getUuid(), body);
            redisClient.expire(req.getUuid(), 5, TimeUnit.MINUTES.name());
            logger.debug("xxxxxxx uuid >>>>  " + req.getUuid() + " 结果存入缓存完成!");
        } else {
            try {
                CallBackModel callBackModel = JSON.parseObject(targetAndEquipment, CallBackModel.class);
                if (callBackModel == null) {
                    redisClient.set(req.getUuid(), body, 5, TimeUnit.MINUTES.name());
                } else {
                    String targetType = callBackModel.getTarget();
                    logger.info("xxxxxxxx targetType >>> " + targetType);
//                    logger.debug("xxxxxxxx equipment >>> " + JSON.toJSONString(callBackModel.getEquipment()));
                    ApiCallBackTarget.Target target = ApiCallBackTarget.getTargetType(targetType);
                    if (target == null) {
                        ///老接口
                        //sendOldMessage(targetType, callBackModel.getEquipment(), body.toString());
                        logger.error("老接口消息，已经弃用，如果看见这个错误，请检查 targetType 是否还在用旧的！！！！");
                    } else {
                        Map<String, Object> messageBody = new HashMap<>(2);
                        messageBody.put("body", body);
                        messageBody.put("equipment", callBackModel.getEquipment());

                        if (target instanceof ApiCallBackTarget.AdPublish) {
                            //广告发布
                            rocketMqProducerService.sendMessage(target.getModuleName(), target.getTargetType(), UUID.get(), messageBody);
                        }  else if (target instanceof ApiCallBackTarget.Robot) {
                            //自動優化營銷目標
                        	rocketMqProducerService.sendMessage(target.getModuleName(), target.getTargetType(), UUID.get(), messageBody);
                        } else if (target instanceof ApiCallBackTarget.AutoControl) {
                            //自动控制
                            handleAutoControllResponse((ApiCallBackTarget.AutoControl) target, messageBody);
                        } else if (target instanceof ApiCallBackTarget.SyncBusiness) {
                            //业务数据同步
                            logger.info("开发发送业务数据: " + target.getModuleName() + " " + "target.getTargetType()");
                            rocketMqProducerService.sendMessage(target.getModuleName(), target.getTargetType(), UUID.get(), this.saveMessageBody2Redis(messageBody));
                            logger.info("结束发送业务数据: " + target.getModuleName() + " " + "target.getTargetType()");
                        }  else if (target instanceof ApiCallBackTarget.CustomAudience) {
                            //业务数据同步
                            rocketMqProducerService.sendMessage(target.getModuleName(), target.getTargetType(), UUID.get(), messageBody);
                        } else if (target instanceof ApiCallBackTarget.Product) {
                            //业务数据同步
                            rocketMqProducerService.sendMessage(target.getModuleName(), target.getTargetType(), UUID.get(), messageBody);
                        }else if (target instanceof ApiCallBackTarget.PublishResource) {
                            //业务数据同步
                            logger.info("开发发送业务数据: " + target.getModuleName() + " " + "target.getTargetType()");
                            rocketMqProducerService.sendMessage(target.getModuleName(), target.getTargetType(), UUID.get(), this.saveMessageBody2Redis(messageBody));
                            logger.info("结束发送业务数据: " + target.getModuleName() + " " + "target.getTargetType()");
                        } else if (target instanceof ApiCallBackTarget.Business) {
                            //业务数据
                            if ((target.getModuleName() + ConfigConst.ROCKETMQ_TOPIC_TAG_SPLIT + "GcByBossCallback").equals(target.getTargetType())) {
                                rocketMqProducerService.sendMessage(target.getModuleName(), target.getTargetType(), UUID.get(), this.saveMessageBody2Redis(messageBody));
                            } else {
                                rocketMqProducerService.sendMessage(target.getModuleName(), target.getTargetType(), UUID.get(), messageBody);
                            }
                        } else if (target instanceof ApiCallBackTarget.InsightsAsyncJob) {
                            ApiCallBackTarget.InsightsAsyncJob realTarget = (ApiCallBackTarget.InsightsAsyncJob) target;
                            switch (realTarget) {
                                case GetInsightsAdGroupAsyncJobData:
                                    rocketMqProducerService.sendMessage(target.getModuleName(), target.getTargetType(), UUID.get(), this.saveMessageBody2Redis(messageBody));
                                    break;
                                default:
                                    rocketMqProducerService.sendMessage(target.getModuleName(), target.getTargetType(), UUID.get(), messageBody);
                            }
                        } else if (target instanceof ApiCallBackTarget.Statitics) {
                            String adSetStatsFetchMes = "produce_" + UUID.get();
                            redisClient.set(adSetStatsFetchMes, body, 20, TimeUnit.MINUTES.name());
                            messageBody.put("body", adSetStatsFetchMes);
                            rocketMqProducerService.sendMessage(target.getModuleName(), target.getTargetType(), UUID.get(), messageBody);
                        } else {
                            rocketMqProducerService.sendMessage(target.getModuleName(), target.getTargetType(), UUID.get(), JSON.toJSONString(messageBody));
                        }
                    }
                }
                logger.debug("xxxxxxx uuid >>>>  " + req.getUuid() + " 结果发布消息队列完成!");
            } catch (Exception e) {
                logger.error("xxxxxxxxxxxxxxxx 处理请求结果异常 uuid >>>>> " + req.getUuid(), e);
            }
        }

    }


    /**
     * 将对象保存到redis中,在使用的时候,根据uuid从redis中获取
     *
     * @param messageBody
     * @return
     */
    private String saveMessageBody2Redis(Map<String, Object> messageBody) {
        String uuid = UUID.get();
        redisClient.set(uuid, messageBody, 3, TimeUnit.HOURS.toString());
        return uuid;
    }

    /**
     * 处理自动控制响应
     *
     * @param target      target
     * @param messageBody messageBody
     */
    private void handleAutoControllResponse(ApiCallBackTarget.AutoControl target, Map<String, Object> messageBody) {
        switch (target) {
            case GetAdSetsStats:
            case GetGroupsStats:
                //统计数据按批次进行消息拆分
                adStatsSplitByBatch((String) messageBody.get("body"), (Map<String, String>) messageBody.get("equipment"), target);
                break;
            case UpdateAdSetBudget:
                rocketMqProducerService.sendMessage(target.getModuleName(), target.getTargetType(), UUID.get(), messageBody);
                break;
            case AdGroupDelete:
            case SyncIns:
            case AdGroupPause:
                rocketMqProducerService.sendMessage(target.getModuleName(), target.getTargetType(), UUID.get(), messageBody);
                break;
            case AdGroupActive:
                rocketMqProducerService.sendMessage(target.getModuleName(), target.getTargetType(), UUID.get(), messageBody);
                break;
            case GetOneGroupsStats:
                rocketMqProducerService.sendMessage(target.getModuleName(), target.getTargetType(), UUID.get(), messageBody);
                break;
            default:
                logger.warn("RedisParamsFBServiceImpl->handleAutoControllResponse->无效的target");
        }
    }

    private void adStatsSplitByBatch(String response, Map<String, String> equipment, ApiCallBackTarget.AutoControl target) {
        String resAdIdMapJson = equipment.get("resourceMap");
        //1、获取adId和批次的对应关系
        Map<String, List<String>> resAdIdMap = JSON.parseObject(resAdIdMapJson, Map.class);

        //2、解析FB返回的响应值，获取统计数据
        List<FBAdStats> fbAdstats;
        try {
            fbAdstats = jsonMapper.toJavaList(response, FBAdStats.class);
        } catch (Exception e) {
            logger.error("RedisParamsFBServiceImpl->adStatsSplitByBatch->解析返回的统计值出现异常->" + response, e);
            return;
        }

        try {
            //3、按批次分组发送消息
            List<FBAdStats> batStats;
            Map<String, Object> messageBody;
            for (Map.Entry<String, List<String>> entry : resAdIdMap.entrySet()) {
                //3-1、构造每个批次对应的统计值
                List<String> adIds = resAdIdMap.get(entry.getKey());
                batStats = new ArrayList<>();
                for (FBAdStats fBAdStats : fbAdstats) {
                    if (ApiCallBackTarget.AutoControl.GetAdSetsStats == target) {
                        if (adIds.contains(fBAdStats.getCampaignId())) {
                            batStats.add(fBAdStats);
                        }
                    } else if (ApiCallBackTarget.AutoControl.GetGroupsStats == target) {
                        if (adIds.contains(fBAdStats.getAdgroupId())) {
                            batStats.add(fBAdStats);
                        }
                    }
                }

                //3-2、构造消息体
                messageBody = new HashMap<>(2);
                messageBody.put("body", JSON.toJSONString(batStats));
                equipment.put("resource", entry.getKey());
                messageBody.put("equipment", equipment);

                //3-3、发送消息
                rocketMqProducerService.sendMessage(target.getModuleName(), target.getTargetType(), UUID.get(),
                        JSON.toJSONString(messageBody));
            }
        } catch (Exception e) {
            logger.error("RedisParamsFBServiceImpl->adStatsSplitByBatch->发送自动控制获取统计值异常", e);
        }
    }


}
